package sink

import java.sql.{Connection, DriverManager, PreparedStatement}
import java.util

import bean.SensorReading
import org.apache.flink.api.scala._
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import source.Source4_UDF.MySensorSource


/**
  * @Description: TODO QQ1667847363
  * @author: xiao kun tai
  * @date:2021/11/13 23:32
  *
  *
  *
  */
object JDBCSink {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    //file source
    val inputPath: String = "src/main/resources/sensor.txt"
    val fileStream: DataStream[String] = env.readTextFile(inputPath)

    //先转换为特定的类型
    val dataStream: DataStream[SensorReading] = fileStream.map(data => {
      val arr = data.split(",")
      SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
    })
    dataStream.print()
    val stream: DataStream[SensorReading] = env.addSource(new MySensorSource)

    stream.addSink(new MyJbdcSinkFunction)

    env.execute("es sink");


  }

  //定义写入一个es的EsSinkFunction
  class MyJbdcSinkFunction() extends RichSinkFunction[SensorReading] {
    //定义连接，预编译语句
    var conn: Connection = _
    var insertStmt: PreparedStatement = _
    var updateStmt: PreparedStatement = _

    override def open(parameters: Configuration): Unit = {
      conn = DriverManager.getConnection("jdbc:mysql://192.168.88.106/flink?characterEncoding=utf8&amp;useSSL=false",
        "root", "000000")
      insertStmt = conn.prepareStatement(
        "insert into sensor_scala (id,temp) values (?,?)");
      updateStmt = conn.prepareStatement(
        "update sensor_scala set temp = ?  where id = ? ");

    }

    override def invoke(value: SensorReading): Unit = {
      //先进行更新操作，查到就更新
      updateStmt.setDouble(1, value.temperature)
      updateStmt.setString(2, value.id)
      updateStmt.execute()
      //如果更新没有查到数据，那么就插入
      if (updateStmt.getUpdateCount == 0) {
        insertStmt.setString(1, value.id)
        insertStmt.setDouble(2, value.temperature)
        insertStmt.execute()
      }
    }

    override def close(): Unit = {

      insertStmt.close()
      updateStmt.close()
      conn.close()
    }
  }

}
